package com.atguigu.gmall.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * @author caodan
 * @version 1.0
 * @date 2022-09-21 10:36
 * <p>
 * 计算用户跳出率
 */
public class DwdTrafficUserJumpDetail {

    public static void main(String[] args) throws Exception {

        //TODO 1 创建flink实时环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        // TODO 设置检查点相关配置
        /*env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // 设置检查点的超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        // 取消job,检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 设置重试
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.seconds(3)));
        // 设置两次检查点的最小间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000);
        // 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs:hadoop102:8020/xxx");*/
        System.setProperty("HADOOP_USER_NAME","atguigu");

        String topicName = "dwd_traffic_page_log";

        String groupId = "dwd_traffic_page_log_group";

        // 获取kafka数据
        FlinkKafkaConsumer<String> flinkKafkaConsumer = KafkaUtil.getFlinkKafkaConsumer(topicName, groupId);

        DataStreamSource<String> kafkaStrDs = env.addSource(flinkKafkaConsumer);

        // 将kafka中的数据jsonStr -> jsonObj
        KeyedStream<JSONObject, String> keyedStream = kafkaStrDs.map(JSON::parseObject)
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps().withTimestampAssigner(
                        new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return element.getLong("ts");
                            }
                        }
                ))
                .keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
        //使用flink CEP 复杂时间从流中获取符合的数据
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("first").where(
                new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject) {
                        String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                        return lastPageId == null;
                    }
                }
        ).next("second").where(
                new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject) {
                        String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                        return lastPageId == null;
                    }
                }
        ).within(org.apache.flink.streaming.api.windowing.time.Time.seconds(10));

        //将pattern应用到流中
        PatternStream<JSONObject> patternStream = CEP.pattern(keyedStream, pattern);

        //定义超时数据侧输出流
        OutputTag<JSONObject> outputTag = new OutputTag<JSONObject>("timeouttag") {
        };
        SingleOutputStreamOperator<JSONObject> flatSelect = patternStream.flatSelect(
                outputTag,
                // 超时数据调用此方法，将数据放入侧输出流中
                new PatternFlatTimeoutFunction<JSONObject, JSONObject>() {
                    @Override
                    public void timeout(Map<String, List<JSONObject>> map, long l, Collector<JSONObject> collector) throws Exception {
                        List<JSONObject> jsonObjectList = map.get("first");
                        for (JSONObject jsonObject : jsonObjectList) {
                            collector.collect(jsonObject);
                        }
                    }
                },
                new PatternFlatSelectFunction<JSONObject, JSONObject>() {
                    @Override
                    public void flatSelect(Map<String, List<JSONObject>> map, Collector<JSONObject> collector) throws Exception {
                        List<JSONObject> jsonObjectList = map.get("first");
                        for (JSONObject jsonObject : jsonObjectList) {
                            collector.collect(jsonObject);
                        }
                    }
                }
        );

        DataStream<JSONObject> sideOutput = flatSelect.getSideOutput(outputTag);
        DataStream<JSONObject> unionDataStream = flatSelect.union(sideOutput);
        SingleOutputStreamOperator<String> operator = unionDataStream.map(JSONAware::toJSONString);
        operator.print(">>>");
        // 写入到kafka中
        operator.addSink(KafkaUtil.getFlinkKafkaProducer("dwd_traffic_user_jump_detail"));
        env.execute();
    }


}
